package org.example.realtime.gongdan.log.job;

import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.example.realtime.gongdan.log.entity.UserLocation;
import org.example.realtime.gongdan.log.function.GaoDeAsyncFunction;

import java.math.BigDecimal;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.util.concurrent.TimeUnit;
/**
 * @Title: UserLocationETL
 * @Author Lianzy
 * @Package org.example.realtime.gongdan.log.job
 * @Date 2025/6/12 14:56
 * @description 过滤源数据：从 dwd_user_behavior 中筛选有经纬度的记录
 * 调用高德API：通过 Async I/O 异步调用接口（避免阻塞）
 * 解析API响应：提取省市区等地理信息
 */
public class UserLocationETL {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql(
                "CREATE TABLE dwd_user_behavior (" +
                        "  user_id BIGINT NOT NULL COMMENT '用户ID'," +
                        "  behavior_type VARCHAR(50) NOT NULL COMMENT '行为类型'," +
                        "  behavior_time TIMESTAMP(3) NOT NULL COMMENT '行为时间'," +
                        "  latitude DECIMAL(10,6) COMMENT '纬度'," +
                        "  longitude DECIMAL(10,6) COMMENT '经度'," +
                        "  ip_address VARCHAR(50) COMMENT 'IP地址'," +
                        "  device_id VARCHAR(100) COMMENT '设备ID'," +
                        "  page_id VARCHAR(100) COMMENT '页面ID'," +
                        "  product_id VARCHAR(50) COMMENT '商品ID'" +
                        ") WITH (" +
                        "  'connector' = 'jdbc'," +
                        "  'url' = 'jdbc:mysql://node101:9030/jtp_data_warehouse'," +
                        "  'table-name' = 'dwd_user_behavior'," +
                        "  'lookup.cache.max-rows' = '1000'," +
                        "  'lookup.cache.ttl' = '1 hour'," +
                        "  'username' = 'root'," +
                        "  'password' = '123456'," +
                        "  'driver' = 'com.mysql.cj.jdbc.Driver'" +
                        ")"
        );

        tableEnv.executeSql(
                "CREATE TABLE dwd_user_location (" +
                        "  user_id BIGINT NOT NULL COMMENT '用户ID（Unique Key）'," +
                        "  province_name VARCHAR(50) COMMENT '省份名称'," +
                        "  city_name VARCHAR(50) COMMENT '城市名称'" +
                        ") WITH (" +
                        "'connector' = 'starrocks',\n" +
                        "'jdbc-url' = 'jdbc:mysql://node101:9030',\n" +
                        "'load-url' = 'node101:8030',\n" +
                        "'database-name' = 'jtp_data_warehouse',\n" +
                        "'table-name' = 'dwd_user_location',\n" +
                        "'username' = 'root',\n" +
                        "'password' = '123456',\n" +
                        "'sink.properties.format' = 'json',\n" +
                        "'sink.properties.strip_outer_array' = 'true'\n" +
                        ")"
        );

        Table sourceTable = tableEnv.sqlQuery(
                "SELECT user_id, latitude, longitude, behavior_time " +
                        "FROM dwd_user_behavior " +
                        "WHERE latitude IS NOT NULL AND longitude IS NOT NULL"
        );
//        sourceTable.execute().print();
        DataStream<UserLocation> inputStream = tableEnv.toDataStream(sourceTable)
                .map(row -> {
                    Long userId = (Long) row.getField("user_id");
                    BigDecimal latitude = (BigDecimal) row.getField("latitude");
                    BigDecimal longitude = (BigDecimal) row.getField("longitude");
                    LocalDateTime behaviorTime = (LocalDateTime) row.getField("behavior_time");
                    Timestamp timestamp = Timestamp.valueOf(behaviorTime);
                    return new UserLocation(userId, latitude, longitude, timestamp);
                });
//        inputStream.print();


        DataStream<UserLocation> resultStream = AsyncDataStream.unorderedWait(
                inputStream,                          // 输入数据流
                new GaoDeAsyncFunction("39a9c562c0d8388c5bbc4c2b31b1bfdb"), // 异步函数实现
                10000,                                // 超时时间（毫秒）
                TimeUnit.MILLISECONDS,                // 时间单位
                10                                    // 并发请求数
        ).name("UserBehaviorSource");
//        resultStream.print();

        tableEnv.createTemporaryView("location_result", resultStream);
//        tableEnv.sqlQuery("select userId,provinceName,cityName from location_result").execute().print();
        tableEnv.executeSql(
                "INSERT INTO dwd_user_location \n" +
                        "SELECT \n" +
                        "  userId AS user_id, \n" +
                        "  provinceName AS province_name, \n" +
                        "  CASE \n" +
                        "    WHEN cityName IS NULL OR cityName = '[]' THEN SUBSTRING(provinceName, 1, 2) \n" +
                        "    ELSE cityName \n" +
                        "  END AS city_name \n" +
                        "FROM location_result"
        );

        env.execute("User Location ETL");
    }
}